package com.aotain.metis.bolt;


import java.util.HashMap;
import java.util.Map;

import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;

import com.aotain.common.CommonFunction;
import com.aotain.apollo.IPScanUpdate;
import com.aotain.storm.AbstractBolt;

/**
 * 
 * IP 活跃实时扫描
 * <功能详细描述>
 * 
 * @author  Turk
 * @version  [版本号, 2017年2月17日]
 * @see  [相关类/方法]
 * @since  [产品/模块版本]
 */
public class IPActiveScanBolt extends AbstractBolt{

	private String dbJson = ""; 
	private HashMap<String,Long> mapIPActive = new HashMap<String,Long>();
	
	private HashMap<String,String> mapIPQQ = new HashMap<String,String>();
	
	/**
	 * 新发现资源
	 */
	private HashMap<String,Long> mapNewIPActive = new HashMap<String,Long>();
	
	/**
	 * serialVersionUID
	 */
	private static final long serialVersionUID = 1L;

	@Override
	public void execute(Tuple tuple) {
		// TODO Auto-generated method stub
		try {
			if(isTickTuple(tuple)) {
				//定时入库
				emitCountingData(collector);
				collector.ack(tuple);
			} else {
				//内存记录IP更新情况
				countInLocal(tuple); 
				collector.ack(tuple);
			}
		} catch (Exception e) {			
			e.printStackTrace();
			collector.fail(tuple);
			throw new FailedException("StatBoltMin throws an exception!",e);
		}
	}
	
	private void countInLocal(Tuple tuple) {
		try
		{
			String ip = tuple.getStringByField("sip").trim();
			
			//判断是否属于局域网IP段范围内
			Long lIP = CommonFunction.ip2int(ip);
			if((lIP >= 167772160 && lIP <= 184549375) 
					|| (lIP >= 2886729728L && lIP <= 2887778303L)
					|| (lIP >= 3232235520L && lIP <= 3232301055L)) {
				long uppkg = tuple.getLongByField("upstreampacket");
				
//				if(ip.equals("192.168.18.40")) {
//					String logInfo = String.format("get IP[%s] uppacket [%d]", ip,uppkg);
//					Logger.getLogger(IPActiveScanBolt.class).info(logInfo);
//				}
				
				if(uppkg <= 10)  //上行包5分钟内大于100个
					return;
				
				if(!mapIPActive.containsKey(ip))
				{
					mapNewIPActive.put(ip, System.currentTimeMillis());
				}
				
				mapIPActive.put(ip, System.currentTimeMillis());
				
				String qq = tuple.getStringByField("qq");
				if(!qq.isEmpty()) {
					mapIPQQ.put(ip, qq);
				}
				
			}
			
			
		} catch (Exception ex) {
			Logger.getLogger(IPActiveScanBolt.class).error("=====countInLocal=========ERROR=================",ex);
		}
		
	}
	
	private void emitCountingData(OutputCollector collector) {
		
		Logger.getLogger(IPActiveScanBolt.class).info("START refrash IP Status,IP num:" + mapIPActive.size());
		Logger.getLogger(IPActiveScanBolt.class).info("METIS UPDATE DATA DBJSON:" + dbJson);
		IPScanUpdate ipUpdate = new IPScanUpdate(dbJson);
		try
		{
			//INSERT
//			for(Entry<String,Long> entry : mapNewIPActive.entrySet()) {
//				ap.Insert(entry.getKey(), entry.getValue());
//			}
			ipUpdate.Insert(mapNewIPActive);
			mapNewIPActive.clear();
			
			//UPDATE
//			for(Entry<String,Long> entry : mapIPActive.entrySet()) {
//				if(entry.getValue() - System.currentTimeMillis() < 6 * 60 * 1000L) {
//					ipUpdate.Update(entry.getKey(), entry.getValue());
//				}
//			}
			ipUpdate.Update(mapIPActive,mapIPQQ);
			
		} catch (Exception ex) {
			Logger.getLogger(IPActiveScanBolt.class).error("==emitCountingData==ERROR====",ex);
		}
	}
	

	@Override
	public void cleanup() {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
		
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,300); //5分钟统计一次
		return conf;
	}

	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		dbJson = stormConf.get("dbJsonMetis").toString(); //oracle
//		Logger.getLogger(IPActiveScanBolt.class).info("METIS DB JSON:" + dbJson);
		//初始化时获取已记录资源数据
		IPScanUpdate ap = new IPScanUpdate(dbJson);
		mapIPActive = ap.getIPList();
	}

}
